package com.example.demo.pool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author cici
 */
@Slf4j
public class TestPool {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(10,2,1,TimeUnit.SECONDS);
        for (int i = 0 ; i < 5 ; i++){
            int j = i;
            threadPool.execute(()->{
                log.info("{}",j);
            });
        }
    }

}

@Slf4j
class ThreadPool{
    private BlockingQueue<Runnable> taskQueue;

    private HashSet<Worker> workers = new HashSet();

    //核心线程数
    private int coreSize;

    private long timeout;

    private TimeUnit timeUnit;

    /**
     * 执行任务
     * @param task
     */
    public void execute(Runnable task){
        //当任务数没有超过coreSize,直接交给worker 对象执行
        //如果任务数超过coreSize,加入任务队列
        synchronized (workers){
            //下面代码非线程安全的，HashSet有查询和加入的操作
            if (workers.size() < coreSize){
                Worker worker = new Worker(task);
                log.info("新增worker{}",worker);
                workers.add(worker);
                worker.start();
            }else {
                log.info("加入任务队列{}",task);
                taskQueue.put(task);
            }
        }
    }

    /**
     * @param queueCapacity  队列长度
     * @param coreSize 核心线程数
     * @param timeout 超时时间
     * @param timeUnit 单位
     */
    public ThreadPool(int queueCapacity, int coreSize, long timeout, TimeUnit timeUnit) {
        this.taskQueue = new BlockingQueue<>(queueCapacity);
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
    }

    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task){
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            //当任务不为空，执行任务
            //task执行完毕，再接着从任务队列获取任务并执行
//            while (task!=null || (task = taskQueue.take()) != null){
            log.info("timeout:{}",timeout);
            log.info("timeUnit:{}",timeUnit);
            while (task != null || (task = taskQueue.poll(timeout,timeUnit)) != null){
                try {
                    log.info("正在执行...{}",task);
                    task.run();
                }catch (Exception e){

                }finally {
                    task = null;
                }
            }
            synchronized (workers){
                log.info("worker 被移除{}",this);
                workers.remove(this);
            }
        }
    }
}


/**
 * 阻塞队列
 * @param <T>
 */
@Slf4j
class BlockingQueue<T>{
    //1、任务队列
    private Deque<T> queue = new ArrayDeque<>();

    //2、锁
    private ReentrantLock lock = new ReentrantLock();

    //3、生产者条件变量
    private Condition fullWaitSet = lock.newCondition();

    //4、消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //5、容量
    private int capacity;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    //带超时的阻塞获取
    public T poll(long timeout, TimeUnit unit){
        lock.lock();
        try{
            long nanos = unit.toNanos(timeout);
            while (queue.isEmpty()){
                try {
                    if (nanos<=0){
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        lock.lock();
        try{
            while (queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T element){
        lock.lock();
        try {
            while (queue.size() == capacity){
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //获取大学
    public int size(){
        lock.lock();
        try {
            return queue.size();
        }finally {
            lock.unlock();
        }
    }
}